package com.galeno.复盘;

import org.apache.commons.lang.RandomStringUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;

/**
 * @author galeno
 * @Title:
 * @Description: 复盘生产者细节
 * @date 2021/9/1310:25
 */
public class GalenoProducer {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        String url = "galeno01:9092,galeno02:9092,galeno03:9092";
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, url);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        /***
         * ask =0 直接发出去,不管是否成功,效率最高,可靠性最差
         * ask =1 发出去,只要有一个leader响应成功,即算发送成功,无需等待follower同步
         * ask=-1或all 必须等待ISR 都完成同步
         */
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "1");
        //最大消息大小,默认为1M
        properties.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "1");
        //压缩方式
        properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        //不包含第一次发送，如果尝试3次都失败，则系统放弃发送 默认是Integer.MAX_VALUE
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, "3");

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            //构造一条消息
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                    "liu",
                    StringUtils.leftPad(i + "", 8, "0"),
                    RandomStringUtils.randomAlphabetic(2).toUpperCase()
            );
            //用生产者发送消息(kafka的producer底层数据发送都是使用的异步发送）
            producer.send(record);
            System.out.println(record);
            Thread.sleep(500);

        }
        producer.close();


    }
}
